package com.cmnit.model

import java.util.Properties

import com.cmnit.utils.{ConfigurationManager, DataSourceUtil, DateUtil, ModelUtils, MySqlUtils}
import org.apache.log4j.Logger
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object GantryLicencePlate {
  val logger: Logger = Logger.getLogger(GantryLicencePlate.getClass)

  def run(sparkSession: SparkSession, gantryFrameBroadCast: Broadcast[DataFrame], vehicleBasicBroadCast: Broadcast[DataFrame], year: String, month: String, day: String, hour: String): Unit = {

    // 获取加工账期（默认当前时间的上1小时）
    val (acctHour, acctDay, acctMonth, acctYear) = ModelUtils.getAcctDate(year, month, day, hour)

    // 拼接当前账期
    val acctTime = acctYear + "-" + acctMonth + "-" + acctDay + " " + acctHour + ":" + "00:00"
    logger.info("当前账期时间：" + acctTime)
    val acctDate = acctYear + acctMonth + acctDay
    // 得到当前账期date对象，计算前一个账期的时间
    val (lastAcctHour, lastAcctDay, lastAcctMonth, lastAcctYear) = ModelUtils.getTime(DateUtil.getTimeByString(acctTime))
    // 拼接前一个账期
    val lastAcctTime = lastAcctYear + "-" + lastAcctMonth + "-" + lastAcctDay + " " + lastAcctHour + ":" + "00:00"
    logger.info("前一个账期时间：" + lastAcctTime)

    // 注册函数
    sparkSession.udf.register("getVlp", (vehicleplate: String) => ModelUtils.getVlpUDF(vehicleplate))
    sparkSession.udf.register("getVlpc", (vehicleplate: String) => ModelUtils.getVlpcUDF(vehicleplate))
    logger.info("注册UDF函数")

    gantryFrameBroadCast.value.createOrReplaceTempView("tmp_organization_gantry")
    logger.info("生成门架中间表:tmp_organization_gantry")
    vehicleBasicBroadCast.value.createOrReplaceTempView("tmp_tbl_vehiclebasicinfo")
    logger.info("生成车牌-车辆类型码表:tmp_tbl_vehiclebasicinfo")

    // 获取最新的门架牌识数据
    sparkSession.sql(
      "select " +
        "* " +
        "from " +
        "ods.ods_etc_gantryvehdisdata " +
        "where vehiclePlate not like '%挂'" +
        " and hour=" + acctHour +
        " and day=" + acctDay +
        " and month=" + acctMonth +
        " and year=" + acctYear)
      .createOrReplaceTempView("temp_etc_gantryvehdisdata")
    logger.info("获取门架牌识数据:temp_etc_gantryvehdisdata")

    // 门架牌识宽表
    sparkSession.sql(
      "select " +
        "t.picId, " +
        "t.gantryId, " +
        "tmp.gantryname, " +
        "tmp.gantrytype, " +
        "tmp.province, " +
        "t.pictime, " +
        "t.driveDir, " +
        "t.cameraNum, " +
        "t.hourBatchNo, " +
        "t.laneNum, " +
        "t.vehiclePlate, " +
        "getVlp(t.vehicleplate) as vlp, " +
        "getVlpc(t.vehicleplate) as vlpc " +
        "from temp_etc_gantryvehdisdata t " +
        "join " +
        "tmp_organization_gantry tmp " +
        "on tmp.gantryid = t.gantryid ")
      .createOrReplaceTempView("tmp_traffic_gantrydisdata_hi")
    logger.info("生成门架牌识宽表:tmp_traffic_gantrydisdata_hi")

    //省界门架流量
    sparkSession.sql(
      "select " +
        "t.province," +
        "t.gantryid," +
        "t.gantrytype," +
        "t.vehicletype," +
        "count(t.vehicleplate) as countnum " +
        "from " +
        "(select " +
        "a.*," +
        "nvl(b.vehicletype, 1) vehicletype " +
        "from " +
        "(select " +
        "distinct " +
        "m.vehicleplate, " +
        "m.province, " +
        "substr(m.gantryid,1,17) as gantryid, " +
        "m.gantryname, " +
        "m.gantrytype " +
        "from " +
        "tmp_traffic_gantrydisdata_hi m) a " +
        "left join " +
        "tmp_tbl_vehiclebasicinfo b " +
        "on a.vehicleplate = b.vehicleplate) t " +
        "group by t.province,t.gantryid,t.gantrytype,t.vehicletype")
      .createOrReplaceTempView("tmp_traffic_province_progantryvecountdisdata_hi")
    logger.info("生成省界门架流量表:tmp_traffic_province_progantryvecountdisdata_hi")

    val prop = new Properties
    prop.setProperty("user", ConfigurationManager.getProperty("mysql.username"))
    prop.setProperty("password", ConfigurationManager.getProperty("mysql.password"))
    prop.setProperty("driver", ConfigurationManager.getProperty("mysql.driver"))

    sparkSession.read
      .jdbc(ConfigurationManager.getProperty("mysql.url"), "traffic_province_gantryvehicletypevecountdisdata_day".toLowerCase, prop)
      .createOrReplaceTempView("tmp_traffic_province_gantryvehicletypevecountdisdata_day")
    logger.info("获取mysql结果表:tmp_traffic_province_gantryvehicletypevecountdisdata_day")

    sparkSession.sql(
      "select " +
        "province, " +
        "gantryid, " +
        "gantrytype, " +
        "vehicletype, " +
        "countnum " +
        "from " +
        "tmp_traffic_province_gantryvehicletypevecountdisdata_day " +
        "where " +
        "date = '" + acctDate + "' " +
        "and time = '" + lastAcctTime + "' ")
      .createOrReplaceTempView("tmp_mysql_gantry")
    logger.info("获取mysql结果表的存量数据:tmp_mysql_gantry")

    // 删除当账期的原数据
    val sql = "delete from traffic_province_gantryvehicletypevecountdisdata_day where time = '" + acctTime + "' and date = '" + acctDate + "'"
    MySqlUtils.execute(sql, DataSourceUtil.getConnection)

    // 省界门架出入省车流量，牌识计算
    sparkSession.sql(
      "select " +
        "nvl(a.province, b.province) province, " +
        "nvl(a.gantryid, b.gantryid) gantryid, " +
        "nvl(a.gantrytype, b.gantrytype) gantrytype, " +
        "nvl(a.vehicletype, b.vehicletype) vehicletype, " +
        "nvl(a.countnum, 0) + nvl(b.countnum, 0) countnum, " +
        "'" + acctTime + "' as time, " +
        "'" + acctDate + "' as date " +
        "from " +
        "tmp_traffic_province_progantryvecountdisdata_hi a " +
        "full join " +
        "tmp_mysql_gantry b " +
        "on " +
        "a.province = b.province " +
        "and a.gantryid = b.gantryid " +
        "and a.gantrytype = b.gantrytype " +
        "and a.vehicletype = b.vehicletype ")
      .write
      .mode(SaveMode.Append)
      .jdbc(ConfigurationManager.getProperty("mysql.url"), "traffic_province_gantryvehicletypevecountdisdata_day".toLowerCase, prop)

    logger.info("省界门架出入省车流量:traffic_province_gantryvehicletypevecountdisdata_day")
    logger.info("插入mysql完毕")
    sparkSession.close
  }
}
